package com.zyb.dfs.datanode.server;

import com.zyb.dfs.namenode.rpc.model.HeartbeatRequest;
import com.zyb.dfs.namenode.rpc.model.HeartbeatResponse;
import com.zyb.dfs.namenode.rpc.model.RegisterRequest;
import com.zyb.dfs.namenode.rpc.model.RegisterResponse;
import com.zyb.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;
import static com.zyb.dfs.datanode.server.DataNodeConfig.*;

import java.util.concurrent.CountDownLatch;

/**
 * 负责跟一组NameNode中的某一个进行通信的线程组件
 *
 * @author zhonghuashishan
 */
public class NameNodeServiceActor {


    private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode;

    public NameNodeServiceActor() {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();
        this.namenode = NameNodeServiceGrpc.newBlockingStub(channel);
    }

    /**
     * 向自己负责通信的那个NameNode进行注册
     */
    public void register(CountDownLatch latch) throws Exception {
        Thread registerThread = new RegisterThread(latch);
        registerThread.start();
        registerThread.join();
    }

    /**
     * 开启发送心跳的线程
     */
    public void startHeartbeat(){
        new HeartbeatThread().start();
    }

    /**
     * 负责注册的线程
     *
     * @author zhonghuashishan
     */
    class RegisterThread extends Thread {

        CountDownLatch latch;

        public RegisterThread(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                // 发送rpc接口调用请求到NameNode去进行注册
                System.out.println("发送rpc请求到NameNode进行注册.......");
                // 当前机器ip、hostname,这两个东西假设是放在配置文件里的
                RegisterRequest request = RegisterRequest.newBuilder()
                        .setIp(DATANODE_IP)
                        .setHostname(DATANODE_HOSTNAME)
                        .setNioPort(NIO_PORT)
                        .build();
                RegisterResponse response = namenode.register(request);
                System.out.println("接收到NameNode返回的注册响应：" + response.getStatus());
                latch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    /**
     * 负责心跳的线程
     *
     * @author zhonghuashishan
     */
    class HeartbeatThread extends Thread {

        @Override
        public void run() {
            try {
                while (true){
                    HeartbeatRequest request = HeartbeatRequest.newBuilder()
                            .setIp(DATANODE_IP)
                            .setHostname(DATANODE_HOSTNAME)
                            .setNioPort(NIO_PORT)
                            .build();
                    HeartbeatResponse response = namenode.heartbeat(request);
                    System.out.println("接收到NameNode返回的心跳响应：" + response.getStatus());

                    Thread.sleep(30 * 1000); // 每隔30秒发送一次心跳到NameNode上去
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

}
